package com.neusoft.bigdata.hbase.biz.impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import com.neusoft.bigdata.hbase.biz.AbstractHBaseBiz;
import com.neusoft.bigdata.hbase.biz.OrderService;
import com.neusoft.bigdata.hbase.model.OrderData;

/**
 * @author lihzh
 * @date 2012-11-21 下午5:44:45
 */
public class OrderServiceHBaseImpl extends AbstractHBaseBiz implements
		OrderService {

	private static HTable orderTable;

	static {
		try {
			orderTable = new HTable(conf, TABLE_NAME_ORDER);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void batchSaveData(List<OrderData> orderdatas) {
		List<Put> puts = createStatisticPuts(orderdatas);
		try {
			orderTable.put(puts);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	private List<Put> createStatisticPuts(List<OrderData> datas) {
		List<Put> puts = new ArrayList<Put>();
		for (OrderData orderdata : datas) {
			String rowkey = createRowKey(orderdata);
			Put put = new Put(Bytes.toBytes(rowkey));
			Map<String, String> valueMap = orderdata.getValueMap();
			Iterator<Entry<String, String>> mapIt = valueMap.entrySet()
					.iterator();
			while (mapIt.hasNext()) {
				Entry<String, String> entry = mapIt.next();
				String name = entry.getKey();
				String value = entry.getValue();
				put.add(Bytes.toBytes(FAMILY_NAME_ORINFO), Bytes.toBytes(name),
						Bytes.toBytes(String.valueOf(value)));
			}
			puts.add(put);
		}
		return puts;
	}

	private String createRowKey(OrderData orderData) {
		StringBuilder strBuilder = new StringBuilder();
		strBuilder.append(orderData.getPrefix()).append("_")
				.append(orderData.getMetircName()).append("_")
				.append(orderData.getCalTime()).append("_")
				.append(orderData.getTimeRange()).append("_")
				.append(orderData.getOrder());
		return createRowKey(orderData.getPrefix().toString(),
				orderData.getMetircName(), orderData.getCalTime(),
				orderData.getTimeRange(), orderData.getOrder());
	}

	private String createRowKey(String prefix, String metricName, long calTime,
			int period, int orderCount) {
		StringBuilder strBuilder = new StringBuilder();
		strBuilder.append(prefix).append("_").append(metricName).append("_")
				.append(calTime).append("_").append(period).append("_")
				.append(orderCount);
		return strBuilder.toString();
	}

	@Override
	public Iterator<Result> searchData(String prefix, String metricName, long calTime,
			int period, int orderCount) {
		String startRowKey = createRowKey(prefix, metricName, calTime, period, 1);
		String endRowKey = createRowKey(prefix, metricName, calTime, period, orderCount);
		try {
			return getByFQN(orderTable, startRowKey, endRowKey);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}
}
